{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import SparkSession\n",
    "from pyspark.sql import SparkSession\n",
    "spark=SparkSession.builder.appName('structured_streaming').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create sample dataset\n",
    "df_1=spark.createDataFrame([(\"XN203\",'FB',300,30),(\"XN201\",'Twitter',10,19),(\"XN202\",'Insta',500,45)], \n",
    "                           [\"user_id\", \"app\" ,\"time_in_secs\",\"age\"]).write.csv(\"demo\",mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#define schema for input data\n",
    "schema=StructType().add(\"user_id\", \"string\").add(\"app\", \"string\").add(\"time_in_secs\", \"integer\").add(\"age\", \"integer\")\n",
    "data=spark.readStream.option(\"sep\", \",\").schema(schema).csv(\"demo\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- user_id: string (nullable = true)\n",
      " |-- app: string (nullable = true)\n",
      " |-- time_in_secs: integer (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_count=data.groupBy('app').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "query=(app_count.writeStream.queryName('count_query').outputMode('complete').format('memory').start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>app</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Insta</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>FB</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       app  count\n",
       "0    Insta      1\n",
       "1       FB      1\n",
       "2  Twitter      1"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from count_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "fb_data=data.filter(data['app']=='FB')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "fb_avg_time=fb_data.groupBy('user_id').agg(F.avg(\"time_in_secs\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "fb_query=(fb_avg_time.writeStream.queryName('fb_query').outputMode('complete').format('memory').start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user_id</th>\n",
       "      <th>avg(time_in_secs)</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>XN203</td>\n",
       "      <td>300.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  user_id  avg(time_in_secs)\n",
       "0   XN203              300.0"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from fb_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_2=spark.createDataFrame([(\"XN203\",'FB',100,30),(\"XN201\",'FB',10,19),(\"XN202\",'FB',2000,45)], \n",
    "                           [\"user_id\", \"app\" ,\"time_in_secs\",\"age\"]).write.csv(\"demo\",mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user_id</th>\n",
       "      <th>avg(time_in_secs)</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>XN203</td>\n",
       "      <td>200.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>XN201</td>\n",
       "      <td>10.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  user_id  avg(time_in_secs)\n",
       "0   XN203              200.0\n",
       "1   XN201               10.0"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from fb_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_3=spark.createDataFrame([(\"XN203\",'FB',500,30),(\"XN201\",'Insta',30,19),(\"XN202\",'Twitter',100,45)], \n",
    "                           [\"user_id\", \"app\" ,\"time_in_secs\",\"age\"]).write.csv(\"demo\",mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user_id</th>\n",
       "      <th>avg(time_in_secs)</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>XN203</td>\n",
       "      <td>200.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>XN201</td>\n",
       "      <td>10.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>XN202</td>\n",
       "      <td>2000.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  user_id  avg(time_in_secs)\n",
       "0   XN203              200.0\n",
       "1   XN201               10.0\n",
       "2   XN202             2000.0"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from fb_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_4=spark.createDataFrame([(\"XN203\",'FB',500,30),(\"XN201\",'Insta',30,19),(\"XN202\",'Twitter',100,45)], \n",
    "                           [\"user_id\", \"app\" ,\"time_in_secs\",\"age\"]).write.csv(\"demo\",mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "#app wise time spent\n",
    "\n",
    "app_df=data.groupBy('app').agg(F.sum('time_in_secs').alias('total_time')).orderBy('total_time',ascending=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_query=(app_df.writeStream.queryName('app_wise_query').outputMode('complete').format('memory').start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>app</th>\n",
       "      <th>total_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>FB</td>\n",
       "      <td>3410</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Insta</td>\n",
       "      <td>560</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>210</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       app  total_time\n",
       "0       FB        3410\n",
       "1    Insta         560\n",
       "2  Twitter         210"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from app_wise_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_5=spark.createDataFrame([(\"XN203\",'FB',500,30),(\"XN201\",'Insta',30,19),(\"XN202\",'Twitter',100,45)], \n",
    "                           [\"user_id\", \"app\" ,\"time_in_secs\",\"age\"]).write.csv(\"csv_folder\",mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>app</th>\n",
       "      <th>total_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>FB</td>\n",
       "      <td>3410</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Insta</td>\n",
       "      <td>560</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>210</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       app  total_time\n",
       "0       FB        3410\n",
       "1    Insta         560\n",
       "2  Twitter         210"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from app_wise_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "# app wise mean age \n",
    "age_df=data.groupBy('app').agg(F.avg('age').alias('mean_age')).orderBy('mean_age',ascending=False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "age_query=(age_df.writeStream.queryName('age_query').outputMode('complete').format('memory').start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>app</th>\n",
       "      <th>mean_age</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>38.500000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>FB</td>\n",
       "      <td>30.571429</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Insta</td>\n",
       "      <td>25.500000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       app   mean_age\n",
       "0  Twitter  38.500000\n",
       "1       FB  30.571429\n",
       "2    Insta  25.500000"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from age_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_6=spark.createDataFrame([(\"XN210\",'FB',500,50),(\"XN255\",'Insta',30,23),(\"XN222\",'Twitter',100,30)], \n",
    "                           [\"user_id\", \"app\" ,\"time_in_secs\",\"age\"]).write.csv(\"csv_folder\",mode='append')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>app</th>\n",
       "      <th>mean_age</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>38.500000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>FB</td>\n",
       "      <td>30.571429</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Insta</td>\n",
       "      <td>25.500000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       app   mean_age\n",
       "0  Twitter  38.500000\n",
       "1       FB  30.571429\n",
       "2    Insta  25.500000"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from age_query \").toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---------+\n",
      "|    app|full_name|\n",
      "+-------+---------+\n",
      "|     FB| FACEBOOK|\n",
      "|  Insta|INSTAGRAM|\n",
      "|Twitter|  TWITTER|\n",
      "+-------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Join static dataframe with streaming dataframe\n",
    "app_df=spark.createDataFrame([('FB','FACEBOOK'),('Insta','INSTAGRAM'),('Twitter','TWITTER')],[\"app\", \"full_name\"])\n",
    "app_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_stream_df=data.join(app_df,'app')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "join_query=(app_stream_df.writeStream.queryName('join_query').outputMode('append').format('memory').start())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>app</th>\n",
       "      <th>user_id</th>\n",
       "      <th>time_in_secs</th>\n",
       "      <th>age</th>\n",
       "      <th>full_name</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>FB</td>\n",
       "      <td>XN201</td>\n",
       "      <td>10</td>\n",
       "      <td>19</td>\n",
       "      <td>FACEBOOK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>FB</td>\n",
       "      <td>XN203</td>\n",
       "      <td>500</td>\n",
       "      <td>30</td>\n",
       "      <td>FACEBOOK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>FB</td>\n",
       "      <td>XN203</td>\n",
       "      <td>500</td>\n",
       "      <td>30</td>\n",
       "      <td>FACEBOOK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>FB</td>\n",
       "      <td>XN203</td>\n",
       "      <td>100</td>\n",
       "      <td>30</td>\n",
       "      <td>FACEBOOK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>FB</td>\n",
       "      <td>XN203</td>\n",
       "      <td>300</td>\n",
       "      <td>30</td>\n",
       "      <td>FACEBOOK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>FB</td>\n",
       "      <td>XN202</td>\n",
       "      <td>2000</td>\n",
       "      <td>45</td>\n",
       "      <td>FACEBOOK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>Insta</td>\n",
       "      <td>XN201</td>\n",
       "      <td>30</td>\n",
       "      <td>19</td>\n",
       "      <td>INSTAGRAM</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>Insta</td>\n",
       "      <td>XN201</td>\n",
       "      <td>30</td>\n",
       "      <td>19</td>\n",
       "      <td>INSTAGRAM</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>Insta</td>\n",
       "      <td>XN202</td>\n",
       "      <td>500</td>\n",
       "      <td>45</td>\n",
       "      <td>INSTAGRAM</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>XN201</td>\n",
       "      <td>10</td>\n",
       "      <td>19</td>\n",
       "      <td>TWITTER</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>XN202</td>\n",
       "      <td>100</td>\n",
       "      <td>45</td>\n",
       "      <td>TWITTER</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>Twitter</td>\n",
       "      <td>XN202</td>\n",
       "      <td>100</td>\n",
       "      <td>45</td>\n",
       "      <td>TWITTER</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        app user_id  time_in_secs  age  full_name\n",
       "0        FB   XN201            10   19   FACEBOOK\n",
       "1        FB   XN203           500   30   FACEBOOK\n",
       "2        FB   XN203           500   30   FACEBOOK\n",
       "3        FB   XN203           100   30   FACEBOOK\n",
       "4        FB   XN203           300   30   FACEBOOK\n",
       "5        FB   XN202          2000   45   FACEBOOK\n",
       "6     Insta   XN201            30   19  INSTAGRAM\n",
       "7     Insta   XN201            30   19  INSTAGRAM\n",
       "8     Insta   XN202           500   45  INSTAGRAM\n",
       "9   Twitter   XN201            10   19    TWITTER\n",
       "10  Twitter   XN202           100   45    TWITTER\n",
       "11  Twitter   XN202           100   45    TWITTER"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"select * from join_query \").toPandas().head(50)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
